package org.future.task.redis.cluster;

import redis.clients.jedis.Client;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.util.JedisClusterCRC16;

import java.util.concurrent.LinkedBlockingQueue;

public class CmdClient extends CommandManager {


    public CmdClient(ClusterFactory clusterFactory, Command command, int commandThreadSize, int localQueueSize) {
        super(clusterFactory, command, commandThreadSize, localQueueSize);
    }

    public CmdClient(ClusterFactory clusterFactory, int commandThreadSize, int localQueueSize) {
        super(clusterFactory, commandThreadSize, localQueueSize);
    }

    public CmdClient(ClusterFactory clusterFactory, Command command, int commandThreadSize) {
        super(clusterFactory, command, commandThreadSize, 0);
    }

    public CmdClient(ClusterFactory clusterFactory, int commandThreadSize) {
        super(clusterFactory, commandThreadSize);
    }


    @Override
    public void putToRedis(CEntry cEntry) throws InterruptedException {
        while (true) {
            //检查
            String bigKey = cEntry.getBigKey();

            int slot = JedisClusterCRC16.getSlot(bigKey);
            String clientName = slotToClient.get(String.valueOf(slot));
            if (clientName == null) {
                while (jedisCluster == null) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                try (Jedis connectionFromSlot = jedisCluster.getConnectionFromSlot(slot); Client client = connectionFromSlot.getClient()) {
                    StringBuilder sb = new StringBuilder();
                    sb.append(client.getHost()).append(":").append(client.getPort());
                    slotToClient.put(String.valueOf(slot), sb.toString());
                    clientName = slotToClient.get(String.valueOf(slot));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            if (clientName == null) {
                Thread.sleep(1);
                continue;
            }
            LinkedBlockingQueue<CEntry> cEntries = clientMapping.get(clientName);
            if (cEntries != null) {
                cEntries.put(cEntry);
                break;
            }
            Thread.sleep(1);
        }
    }

    @Override
    public void syncPutToRedis(CEntry cEntry) {
        //检查
        String bigKey = cEntry.getBigKey();

        int slot = JedisClusterCRC16.getSlot(bigKey);
        String clientName = slotToClient.get(String.valueOf(slot));
        if (clientName == null) {
            while (jedisCluster == null) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            try (Jedis connectionFromSlot = jedisCluster.getConnectionFromSlot(slot)) {
                try (Pipeline pipelined = connectionFromSlot.pipelined()) {
                    command.exec(pipelined, cEntry);
                    pipelined.sync();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
